/*
 * Based on:
 *
 * balance - a balancing tcp proxy
 * $Revision: 3.42 $
 *
 * Copyright (c) 2000-2007,2008 by Thomas Obermair (obermair@acm.org)
 * and Inlab Software GmbH (info@inlab.de), Gruenwald, Germany.
 * All rights reserved.
 */

#include <balance.h>
#include <tanger-stm.h>

COMMON *common;

static int hashfailover = 0;
static int debugflag = 0;
static int foreground = 0;
static int packetdump = 0;
static int sockbufsize = SOCKET_BUFFER_SIZE;
static char *balance_socket_path = NULL;
static struct timeval sel_tmout = { 0, 0 }; /* seconds, microseconds */
static struct timeval save_tmout = { 0, 0 }; /* seconds, microseconds */

int max_retries = 10;
char workers_path[FILE_NAME_LEN];

unsigned long client_connects = 0;


void printStatistics(){
	fprintf(stderr, "\nnumber of clients %d \n", client_connects);
	fprintf(stderr, "number of failures %d \n", cmn_fc(common));
	int i;
	fprintf(stderr, "number of groups %d \n", cmn_ngroups(common));
	for(i = 0; i < cmn_ngroups(common); i++){
		int j;
		fprintf(stderr, "\tnumber of channels %d \n", grp_nchannels(common,i));
		for(j = 0; j < grp_nchannels(common,i); j++) {
			fprintf(stderr, "\t\tgroup:%d channel:%d tc: %d, bsent: %d bresv: %d \n",
					i,j, chn_tc(common, i,j), chn_bsent(common, i,j), chn_breceived(common, i,j));
		}
	}
	fflush(stderr);
}
/*
 * to print statistics
 * ps -x | grep balance
 * kill -s sigusr1 'pid'
 */
void sig_handler(int sig)
{
	signal(sig, sig_handler); /* reset it to this function */
    printStatistics();
}

static int fatal_err_dump(char *text) {
	printStatistics();
	perror(text);
	tanger_thread_shutdown();
	tanger_shutdown();
	free(common);
	exit(-1);
}

static unsigned int hash_fold(char* s, int len) {
	unsigned int rc = 0;
	int i = 0;
	for (i = 0; i < len; i++) {
		rc = s[i] + 31 * rc;
	}
	return (rc);
}

static ssize_t writen(int fd, unsigned char *ptr, size_t nbytes) {
	int nleft;
	ssize_t nwritten;

	nleft = nbytes;

	while (nleft > 0) {
		nwritten = write(fd, ptr, nleft);
		if (nwritten <= 0) {
			return (nwritten); /* error */
		}
		nleft -= nwritten;
		ptr += nwritten;
	}

	return (nbytes - nleft);
}



int init_serversocket(char* socket_path) {
	int srv_socket, status, sockopton;
	struct sockaddr_un server;
	int len;

	srv_socket = socket(AF_UNIX, SOCK_STREAM, 0);
	if (srv_socket < 0) {
		fatal_err_dump("socket()");
	}

	/* configure balance's socket, make it reusable and define buffers' length in both directions*/
	//solves "address already in use"
	sockopton = 1;
	setsockopt(srv_socket, SOL_SOCKET, SO_REUSEADDR, (char*) &sockopton, sizeof(sockopton));
	//setsockopt(srv_socket, SOL_SOCKET, SO_SNDBUF, &sockbufsize, sizeof(sockbufsize));
	//setsockopt(srv_socket, SOL_SOCKET, SO_RCVBUF, &sockbufsize, sizeof(sockbufsize));

	server.sun_family = AF_UNIX;
	strcpy(server.sun_path, socket_path);
	unlink(server.sun_path);
	len = strlen(server.sun_path) + sizeof(server.sun_family);

	status = bind(srv_socket, (struct sockaddr *)&server, len);
	if (status < 0) {
		fatal_err_dump("bind()");
	}

	status = listen(srv_socket, BALANCE_QUEUE_LENGTH);
	if (status < 0) {
		fatal_err_dump("listen()");
	}
	return (srv_socket);
}

void init_worker_channel(char* chanel_path, int *maxc, char* argv_i) {
	char *p;
	/* handle unix socket path*/
	if((p = strtok (argv_i,":")) == NULL){
		fatal_err_dump("worker unix socket path should be specified");
	}else{
		strcpy(chanel_path, workers_path);
		strcat(chanel_path, p);
	}
	/* handle max connection for worker*/
	if((p = strtok (NULL, ":")) == NULL){
		*maxc = 0;
	}else{
		*maxc = atoi(p);
	}
	if(debugflag){
		fprintf(stderr, "worker: %s,%d\n", chanel_path, *maxc);
	}

	if(debugflag){
		struct sockaddr_un remote;
		int s;
		if ((s = socket(AF_UNIX, SOCK_STREAM, 0)) == -1) {
			perror("socket error to worker");
			exit(1);
		}

		remote.sun_family = AF_UNIX;
		strcpy(remote.sun_path, chanel_path);
		int len = strlen(remote.sun_path) + sizeof(remote.sun_family);
		if (connect(s, (struct sockaddr *)&remote, len) == -1) {
			fprintf(stderr, "connect to worker %s failed\n", remote.sun_path);
			perror("connect error");
			exit(1);
		}
		else{
			close(s);
		}
	}
}

int balance(int* last_group_index, struct in_addr* cli_addr, enum boolean_t from_master) {
	/* index over groups */
	int start_group_index = *last_group_index;
	int current_group_index = start_group_index;
	/* index over channels in group */
	int current_channel = -1;
	/* error code for transaction*/
	int error_code = 0;

	int cicle_counter = 0;

	tanger_begin();
	//for over groups
	for (;;) {
		//start finding place in current group from next channel to current one or from 0 if it's first time
		if(current_group_index == -1){
			//supposing there is always 0 group present with one channel
			current_group_index = 0;
			current_channel = 0;
		}else{
			//try next channel in same group
			current_channel = (grp_current(common, current_group_index) + 1) % grp_nchannels(common, current_group_index);
		}

		//for over channels in group
		for (;;) {
			if (grp_type(common, current_group_index) == GROUP_RR) {
				/* if there is place in current channel take it*/
				if (chn_status(common, current_group_index, current_channel) == CHANNEL_ENABLED &&
						(chn_maxc(common, current_group_index, current_channel) == 0 || (chn_c(common, current_group_index, current_channel)< chn_maxc(common, current_group_index, current_channel)))) {
					break;
				}
				//move to next channel if current is full
				else {
					current_channel++;
					//mod # of channels in group
					if (current_channel >= grp_nchannels(common, current_group_index)) {
						current_channel = 0;
					}
					//we made full cicle
					if (current_channel == grp_current(common, current_group_index)) {
						current_channel = -1; // no channel available in this group
						break;
					}
				}

			} else if (grp_type(common,current_group_index) == GROUP_HASH) {
				unsigned int uindex = hash_fold((char *) cli_addr, sizeof(*cli_addr));
				current_channel = uindex % grp_nchannels(common, current_group_index);
				//if there is place in current channel
				if (chn_status(common, current_group_index, current_channel) == CHANNEL_ENABLED &&
						(chn_maxc(common, current_group_index, current_channel) == 0 || (chn_c(common, current_group_index, current_channel) < chn_maxc(common, current_group_index, current_channel)))) {
					break;
				}//choose another group
				else {
					//hash failover is done by RR technique from current_channel in current group
					if (hashfailover == 1) {
						int startindex = current_channel;
						for (;;) {
							current_channel++;
							//mod
							if (current_channel >= grp_nchannels(common, current_group_index)) {
								current_channel = 0;
							}
							//made whole cicle in current group
							if (current_channel == startindex) {
								current_channel = -1;
								break;
							}
							//check if channel isn't full
							if (chn_status(common, current_group_index, current_channel) == CHANNEL_ENABLED &&
									(chn_maxc(common, current_group_index, current_channel) == 0 || (chn_c(common, current_group_index, current_channel) < chn_maxc(common, current_group_index, current_channel)))) {
								break;
							}
						}
					} else {
						current_channel = -1;
					}
					break;
				}
			}//wrong group type
			else {
				error_code = 1;
				break;
			}
		}//for over channels

		if (current_channel < 0) {
			// try next group starting from 0 channel
			current_group_index = (current_group_index + 1) % cmn_ngroups(common);
			current_channel = 0;

			if(current_group_index == start_group_index){
				if(cicle_counter==max_retries){
//					if(debugflag){
//						fprintf(stderr, "start_group, current_group, current_channel %d %d %d", start_group_index, current_group_index, current_channel);
//					}
					current_group_index = -1;
					current_channel = -1;
					break;
				}else{
					cicle_counter++;
					continue;
				}
			}
		}/* we found legal channel in current group */
		else{
			break;
		}
	}//for over groups

	/* all groups and all channels are full*/
	if (current_channel < 0 || current_group_index < 0) {
		cmn_fc(common)++;
		error_code = 2;
	}
	/* we found place in current group */
	else{
		//remember group where we find empty channel so next time start from it
		if(from_master == TRUE){
			*last_group_index = current_group_index;
		}

		//remember current channel in group
		grp_current(common, current_group_index) = current_channel;

		//reserve channel for client(still unsuccessful connection is possible)
		chn_c(common, current_group_index, current_channel)++;
		chn_tc(common, current_group_index, current_channel)++;

	}
	tanger_commit();

	switch (error_code){
		//tanger_thread_shutdown();
		//tanger_shutdown();
	case 1:
		fatal_err_dump("PANIC: invalid group type \n");
		break;
	case 2:
		if(debugflag){
			fprintf(stderr, "all groups and all channels are full \n");
		}
		break;
	}

	return current_channel;
}//end of balance

void update_bytes_on_channel(int groupindex, int channelindex, ssize_t rc, enum boolean_t is_forward) {
	tanger_begin();
	if(is_forward == TRUE){
		chn_bsent(common, groupindex, channelindex) += rc;
	}else{
		chn_breceived(common, groupindex, channelindex) += rc;
	}
	tanger_commit();
}

int relay(int fromfd, int tofd, int groupindex, int channelindex, enum boolean_t is_forward) {
	ssize_t rc;
	char* print_symbol = (is_forward) ? "->" : "<-";
	unsigned char buffer[MAX_MESSAGE_SIZE];
	bzero(buffer,MAX_MESSAGE_SIZE);
	rc = read(fromfd, buffer, MAX_MESSAGE_SIZE);

	//we ended forwarding data
	if (rc <= 0) {
		return (-1);
	} else {
		if (packetdump) {
			fprintf(stderr, "%s %s, %d\n", print_symbol, buffer, rc);
		}
		if (writen(tofd, buffer, rc) != rc) {
			return (-1);
		}
		update_bytes_on_channel(groupindex, channelindex, rc, is_forward);
	}
	return (0);
}



/*
 * the connection is really established, let's transfer the data
 *  as efficient as possible :-)
 */
void relay_stream(int clientfd, int worker_socketfd, int group_index, int channel_index) {
	//client_worker_fds will be modified to reflect which of the file descriptors you selected which is ready for reading
	fd_set client_worker_fds;
    fd_set read_fds;  // temp file descriptor list for select()
	int fdset_width;
	int sr;
	int optone = 1;

	//The parameter fdset_width should be set to the values of the highest file descriptor plus one
	fdset_width = ((clientfd > worker_socketfd) ? clientfd : worker_socketfd) + 1;

	/* before relaying configure woker's socket*/
	//setsockopt(worker_socketfd, IPPROTO_TCP, TCP_NODELAY, (char *) &optone, (socklen_t) sizeof(optone));
	//setsockopt(worker_socketfd, SOL_SOCKET, SO_KEEPALIVE, (char *) &optone, (socklen_t) sizeof(optone));
	//configure worker socket
	//setsockopt(worker_sockfd, SOL_SOCKET, SO_SNDBUF, &sockbufsize, sizeof(sockbufsize));
	//setsockopt(worker_sockfd, SOL_SOCKET, SO_RCVBUF, &sockbufsize, sizeof(sockbufsize));

	/* configure client's socket */
	//setsockopt(clientfd, IPPROTO_TCP, TCP_NODELAY, (char *) &optone, (socklen_t) sizeof(optone));
	//setsockopt(clientfd, SOL_SOCKET, SO_KEEPALIVE, (char *) &optone, (socklen_t) sizeof(optone));

	//Clear all entries from the set
	FD_ZERO(&client_worker_fds);
	FD_ZERO(&read_fds);
	//Add clientfd to the set
	FD_SET(clientfd, &client_worker_fds);
	//Add workerfd to the set
	FD_SET(worker_socketfd, &client_worker_fds);

	//relay for
	for (;;) {
		/*
		 * just in case this system modifies the timeout values or previous select modifies read_fds,
		 * refresh the values from a saved copy of them.
		 */
		read_fds = client_worker_fds;
		sel_tmout = save_tmout;

		//select on open socket
		for (;;) {

			if (sel_tmout.tv_sec || sel_tmout.tv_usec) {
				sr = select(fdset_width, &read_fds, NULL, NULL, &sel_tmout);
			} else {
				sr = select(fdset_width, &read_fds, NULL, NULL, NULL);
			}
			if ((save_tmout.tv_sec || save_tmout.tv_usec) && !sr) {
				fprintf(stderr, "timed out after %d seconds\n", (int) save_tmout.tv_sec);
				fflush(stderr);
				exit(EX_UNAVAILABLE);
			}
			//EINTR interuption by some signal
			if (sr < 0 && errno != EINTR) {
				fatal_err_dump("select error");
			}
			/* something present over sockets*/
			if (sr > 0){
				break;
			}
		}//end of for select open client/worker sockets

		//if client sent data forward it to worker
		if (FD_ISSET(clientfd, &read_fds)) {
			if (relay(clientfd, worker_socketfd, group_index, channel_index, TRUE) < 0) {
				break;
			}
		}//otherwise backward data from worker to client
		if (FD_ISSET(worker_socketfd, &read_fds)) {
			if (relay(worker_socketfd, clientfd, group_index, channel_index, FALSE) < 0) {
				break;
			}
		}
	}//end of relay for
}


/*
 * tries to connect to selected channel in group
 * after successful connection delegates stream relay to stream_relay function
 */
void* handle_client(void* client_param) {
		thread_params_t* params = (thread_params_t*) client_param;
		int clientfd = params->clientfd;
		int group_index = params->group_index;
		int channel_index = params->channel_index;
		struct in_addr client_address =  params->client_address;
		enum boolean_t is_one_worker = params->is_one_worker;
		free(client_param);

		struct sockaddr_un worker_addr;
		worker_addr.sun_family = AF_UNIX;

		/* main thread already initilized tanger */
		if(is_one_worker  ==  FALSE){
			tanger_thread_init();
		}

		if(debugflag){
			fprintf(stderr, "working with group:channel %d:%d", group_index, channel_index);
		}
		int worker_sockfd = 0;
		//connection for to worker server
		for (;;) {
			if (debugflag) {
				fprintf(stderr, "trying to connect to worker in group %d channel %d \n", group_index, channel_index);
			}
			if ((worker_sockfd = socket(AF_UNIX, SOCK_STREAM, 0)) < 0) {
				fatal_err_dump("can't open stream socket");
			}

			//no need to synch, since chn_path is read-only after init
			strcpy(worker_addr.sun_path, chn_path(common, group_index, channel_index));

			if (connect(worker_sockfd, (struct sockaddr *) &worker_addr, sizeof(worker_addr)) < 0) {
				close(worker_sockfd);
				if (errno == EINTR) {
					fprintf(stderr, "timeout group %d channel %d\n", group_index, channel_index);
				} else {
					fprintf(stderr, "connection refused group %d channel %d\n",	group_index, channel_index);
				}

				/* reservation broken, clean the channel */
				tanger_begin();
				chn_c(common, group_index, channel_index)--;
				chn_tc(common, group_index, channel_index)--;
				tanger_commit();

				//find another group and channel
				fprintf(stderr, "rebalancing..");
				if((channel_index = balance(&group_index, &client_address, FALSE))> 0){
					continue;
				}else{
					close(clientfd);
					if(is_one_worker == FALSE){
						pthread_exit(0);
					}else{
						return;
					}
				}
			}
			// we made successful connection to worker server
			else {
				if (debugflag) {
					fprintf(stderr, "successful connect on reserved group %d, channel %d \n",group_index, channel_index);
				}
				//reservation is correct, connection succeeded, proceed to relay streams
				relay_stream(clientfd, worker_sockfd, group_index, channel_index);
				break;
			}
		}//end of connection for to worker server

		/* after all work done free the channel*/
		if(debugflag){
			fprintf(stderr, "releasing group:channel %d:%d \n ", group_index, channel_index);
		}
		tanger_begin();
		chn_c(common, group_index, channel_index)--;
		tanger_commit();

		close(worker_sockfd);
		close(clientfd);

		/* no need to terminate main thread */
		if(is_one_worker  == FALSE ){
			if(debugflag){
				fprintf(stderr, "<<<<<<closing thread\n");
			}
			tanger_thread_shutdown();
			pthread_exit(0);
		}else{
			return;
		}
}


/*
 *For some reason tanger can't handle conditional tanger_thread_init(), probably bug in gc_init_thread:gc.c:tinySTM?
 * */
void* handle_one_client(void* client_param) {
		thread_params_t* params = (thread_params_t*) client_param;
		int clientfd = params->clientfd;
		int group_index = params->group_index;
		int channel_index = params->channel_index;
		struct in_addr client_address =  params->client_address;
		free(client_param);

		struct sockaddr_un worker_addr;
		worker_addr.sun_family = AF_UNIX;

		if(debugflag){
			fprintf(stderr, "working with group:channel %d:%d", group_index, channel_index);
		}
		int worker_sockfd = 0;
		//connection for to worker server
		for (;;) {
			if (debugflag) {
				fprintf(stderr, "trying to connect to worker in group %d channel %d \n", group_index, channel_index);
			}
			if ((worker_sockfd = socket(AF_UNIX, SOCK_STREAM, 0)) < 0) {
				fatal_err_dump("can't open stream socket");
			}

			//no need to synch, since chn_path is read-only after init
			strcpy(worker_addr.sun_path, chn_path(common, group_index, channel_index));

			if (connect(worker_sockfd, (struct sockaddr *) &worker_addr, sizeof(worker_addr)) < 0) {
				close(worker_sockfd);
				if (errno == EINTR) {
					fprintf(stderr, "timeout group %d channel %d\n", group_index, channel_index);
				} else {
					fprintf(stderr, "connection refused group %d channel %d\n",	group_index, channel_index);
				}

				/* reservation broken, clean the channel */
				tanger_begin();
				chn_c(common, group_index, channel_index)--;
				chn_tc(common, group_index, channel_index)--;
				tanger_commit();

				//find another group and channel
				fprintf(stderr, "rebalancing..");
				if((channel_index = balance(&group_index, &client_address, FALSE))> 0){
					continue;
				}else{
					close(clientfd);
					return;
				}
			}
			// we made successful connection to worker server
			else {
				if (debugflag) {
					fprintf(stderr, "successful connect on reserved group %d, channel %d \n",group_index, channel_index);
				}
				//reservation is correct, connection succeeded, proceed to relay streams
				relay_stream(clientfd, worker_sockfd, group_index, channel_index);
				break;
			}
		}//end of connection for to worker server

		/* after all work done free the channel*/
		if(debugflag){
			fprintf(stderr, "releasing group:channel %d:%d \n ", group_index, channel_index);
		}
		tanger_begin();
		chn_c(common, group_index, channel_index)--;
		tanger_commit();

		close(worker_sockfd);
		close(clientfd);

		return;
}

static void usage(void) {
	fprintf(stderr, "  usage:\n");
	fprintf(stderr,	"  balance [-b balance-unix-socket-path] [-t sec] [-T sec] [-dfpH] \\\n");
	fprintf(stderr,	"          [path[:maxc1]] [!%%] [ ... hN[:pN[:maxcN]]]]\n");
	fprintf(stderr, "\n");
	fprintf(stderr, "  -c #cpus:cpu to bind to\n");
	fprintf(stderr, "  -s #path to workers directory\n");
	fprintf(stderr, "  -b path   bind to specific socket-path to listen\n");
	fprintf(stderr, "  -d        debugging on\n");
	fprintf(stderr, "  -f        stay in foregound\n");
	fprintf(stderr, "  -H        failover even if Hash Type is used\n");
	fprintf(stderr, "  -p        packet dump \n");
	fprintf(stderr,	"  -T sec    timeout (seconds) for select (0 => never) (default=%d)\n", DEFAULT_SELECT_TIMEOUT);
	fprintf(stderr,	"   !        separates channel groups (declaring previous to be Round Robin)\n");
	fprintf(stderr,	"   %       as !, but declaring previous group to be a Hash Type\n");
	fprintf(stderr, "\n");
	fprintf(stderr, "example:\n");
	fprintf(stderr, "balance -b ./sockets/balance -t 10 -T 10 -d -f -p -H ./sockets/w1:10 ./sockets/w2 ! ./sockets/w3:100 %% \n");
	fprintf(stderr, "\n");

	exit(EX_USAGE);
}
// goto background:

void background(void) {
	int childpid;
	if ((childpid = fork()) < 0) {
		fprintf(stderr, "cannot fork\n");
		exit(EX_OSERR);
	} else {
		if (childpid > 0) {
			exit(EX_OK); /* parent */
		}
	}
	setpgrp();
	chdir("/");
	close(0);
	close(1);
	close(2);
}

/* done in main thread, no need to synch*/
COMMON *makecommon(int argc, char **argv) {
	int i;
	int group;
	int channel;
	COMMON *mycommon;

	if ((mycommon = (COMMON *) malloc(sizeof(COMMON))) == NULL) {
		fprintf(stderr, "cannot allocate COMMON struct\n");
		exit(EX_OSERR);
	}

	mycommon->fc = 0;
	mycommon->pid = getpid();

	fprintf(stderr,  "pid: %d \n", mycommon->pid);

	for (group = 0; group < MAX_GROUPS; group++) {
		grp_nchannels(mycommon, group) = 0;
		grp_current(mycommon, group) = 0;
		grp_type(mycommon, group) = GROUP_RR; // Default: RR
	}

	group = 0;
	channel = 0;

	for (i = 0; i < argc; i++) {
		/*end of round robin group*/
		if (strcmp(argv[i], "!") == 0) {
			if (channel <= 0) {
				fatal_err_dump("no channels in group");
			}
			grp_type(mycommon, group) = GROUP_RR;
			grp_nchannels(mycommon, group) = channel;

			/* proceed to next group*/
			group++;
			channel = 0;
			if (group >= MAX_GROUPS) {
				fatal_err_dump("too many groups");
			}
		}/* end of hash group*/
		else if (strcmp(argv[i], "%") == 0) {
			if (channel <= 0) {
				fatal_err_dump("no channels in group");
			}
			grp_type(mycommon, group) = GROUP_HASH;
			grp_nchannels(mycommon, group) = channel;

			/* proceed to next group*/
			group++;
			channel = 0;
			if (group >= MAX_GROUPS) {
				fatal_err_dump("too many groups");
			}
		} else {
			chn_status(mycommon, group, channel) = CHANNEL_ENABLED;
			chn_c(mycommon, group, channel) = 0;    // connections...
			chn_tc(mycommon, group, channel) = 0;   // total connections...
			init_worker_channel(chn_path(mycommon, group, channel), &chn_maxc(mycommon, group, channel), argv[i]);
			chn_bsent(mycommon, group, channel) = 0;
			chn_breceived(mycommon, group, channel) = 0;

			channel++;
			if (channel >= MAX_CHANNELS) {
				fatal_err_dump("too many channels in one group");
			}
		}
	}
	/* in a case user didn't give group type */
	if(channel > 0){
		grp_type(mycommon, group) = GROUP_RR;
		grp_nchannels(mycommon, group) = channel;
		group++;
	}
	cmn_ngroups(mycommon) = group;

	if (debugflag) {
		fprintf(stderr, "number of active groups: %d\n", cmn_ngroups(mycommon));
		for (group = 0; group < cmn_ngroups(mycommon); group++) {
			fprintf(stderr, "number of channels %d \n", grp_nchannels(mycommon, group));
			for (i = 0; i < grp_nchannels(mycommon, group); i++) {
				fprintf(stderr, "%d %d %s:%d\n", group,	i, chn_path(mycommon, group, i), chn_maxc(mycommon, group, i));
				fflush(stderr);
			}
		}
	}

	return (mycommon);
}
void handle_affinity(char* cpu_arg){
		int pid = getpid();
		if (cpu_arg != NULL) {
		fprintf(stderr, "cpu arg %s\n", cpu_arg);
		cpu_set_t* cpu_set;
		char* p = NULL;
		int num_of_cpus = 0;
		if ((p = strtok(cpu_arg, ":")) == NULL) {
			perror("split error");
			return;
		} else {
			num_of_cpus = atoi(p);
			fprintf(stderr, "num of cpus %d\n", num_of_cpus);
			cpu_set = CPU_ALLOC(num_of_cpus);
		}
		size_t size = CPU_ALLOC_SIZE(num_of_cpus);

		CPU_ZERO_S(size, cpu_set);
		while ((p = strtok(NULL, ",")) != NULL) {
			int cpu_num = atoi(p);
			fprintf(stderr, "cpu_num to run on %d\n", cpu_num);
			CPU_SET_S(cpu_num, size, cpu_set);
		}

		fflush(stderr);

		if (sched_setaffinity(pid, size, cpu_set) < 0) {
			perror("sched_setaffinity");
		}

		if (sched_getaffinity(pid, size, cpu_set) < 0) {
			perror("sched_getaffinity");
			return -1;
		}
		fprintf(stderr, "pid %d's affinity mask: %d\n", pid, *cpu_set);
		CPU_FREE(cpu_set);
	}
}

int main(int argc, char *argv[]) {
	/* used to parse options*/
	int c;
	enum boolean_t is_one_worker = FALSE;
	while ((c = getopt(argc, argv, "c:b:T:s:dfpHo")) != EOF) {
			switch (c) {
			case 'c':
				handle_affinity(optarg);
				break;
			case 'b':
				balance_socket_path = optarg;
				break;
			case 'T':
				sel_tmout.tv_sec = atoi(optarg);
				sel_tmout.tv_usec = 0;
				if (sel_tmout.tv_sec < 1)
					usage();
				save_tmout = sel_tmout;
				break;
			case 's':
				strcpy(workers_path,optarg);
				break;
			case 'f':
				foreground = 1;
				break;
			case 'd':
				debugflag = 1;
				break;
			case 'p':
				packetdump = 1;
				break;
			case 'H':
				hashfailover = 1;
				break;
			case 'o':
				is_one_worker = TRUE;
				break;
			case '?':
			default:
				usage();
			}
		}

	argc -= optind;
	argv += optind;


	fprintf(stderr,"balance_socket_path %s : sel_tmout.tv_sec %ld :foreground %d : debugflag %d : packetdump %d : hashfailover %d \n",
			balance_socket_path,sel_tmout.tv_sec,foreground,debugflag,packetdump,hashfailover);
	fprintf(stderr, "is one worker : %d \n", is_one_worker);
	if (debugflag) {
		fprintf(stderr, "after parsing options: \n");
		fprintf(stderr, "argv[0]=%s\n", argv[0]);
		fprintf(stderr, "argc : %d\n", argc);
	}

	if (!foreground) {
		background();
	}

	if (signal(SIGUSR1, sig_handler) == SIG_ERR) {
        perror("signal");
        exit(1);
    }

	// Open a server's UNIX socket
	int sockfd = init_serversocket(balance_socket_path);

	/* create shared groups'/channels' data*/
	common = makecommon(argc, argv);

	/* start balancing from 0 group, multi-reader var */
	int last_group_index = -1;

	pthread_attr_t attr;
	pthread_attr_init(&attr);
	pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);

	tanger_init();
	tanger_thread_init();

	for (;;) {
		char client_id_str[SIZE_OF_ID];
		bzero(client_id_str, SIZE_OF_ID);


		struct sockaddr_un cli_addr;
		/* length of client socket structure*/
		int clilen = sizeof(cli_addr);
		int new_client_socket = accept(sockfd, (struct sockaddr *) &cli_addr, &clilen);
		client_connects++;
		if (new_client_socket < 0) {
			fatal_err_dump("error in client socket creation");
		}
		/* first INET_ADDRSTRLEN bytes of packet is id(or ip) of client */
		int n = recv(new_client_socket, (&client_id_str), SIZE_OF_ID, 0);
		if (n <= 0) {
                        perror("recv error, id isn't sent");
                        close(new_client_socket);
                        continue;

		}

		/* client address for local unix-socket*/
		struct in_addr client_id;
		if(inet_aton(client_id_str, &client_id) < 0){
			fatal_err_dump("inet_ntoa error");
		}
		if(client_id.s_addr < 0 ){
			fatal_err_dump("illegal client id");
		}
		if (debugflag) {
			fprintf(stderr, ">>>>>>new connection from %s read n: %d bytes \n", client_id_str, n);
			u_int32_t ip = client_id.s_addr;
			fprintf(stderr, "%d.%d.%d.%d\n", (ip >> 24) & 0xFF, (ip >> 16) & 0xFF, (ip >> 8) & 0xFF, ip & 0xFF);
		}

		int allocated_channel = -1;
		/* balance from group where place was last time found*/
		if ((allocated_channel = balance(&last_group_index, &client_id, TRUE)) >= 0) {
			if (debugflag) {
				fprintf(stderr, "allocated %d group and %d channel\n", last_group_index, allocated_channel);
			}
			pthread_t thread;
			thread_params_t* params = (thread_params_t*) malloc(sizeof(thread_params_t));
			params->clientfd = new_client_socket;
			params->group_index = last_group_index;
			params->channel_index = allocated_channel;
			params->client_address = client_id;
			params->is_one_worker = is_one_worker;

			if(is_one_worker == FALSE){
				if(debugflag){
					fprintf(stderr, "creating thread");
				}
				if (pthread_create(&thread, &attr, handle_client, (void *) params) != 0) {
					fatal_err_dump("Error creating thread\n");
				}
			}else{
				handle_one_client(params);
			}

		}else{
			close(new_client_socket);
		}
	}//end of accept for

	pthread_attr_destroy(&attr);
	tanger_thread_shutdown();
	tanger_shutdown();
	free(common);
}
